package com.lagou.java.druid;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
import org.apache.flink.util.Collector;

import java.util.Properties;

public class Transform {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        String topic = "lagoudruid2";
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers","node51:9092,node52:9092,node53:9092");
        prop.setProperty("group.od","local");

        DataStream<String> json = env.addSource(new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),prop));

        SingleOutputStreamOperator<OrderOrigin> singleOutputStreamOperator = json.map((MapFunction<String, OrderOrigin>) s -> {
            OrderOrigin order = JSON.parseObject(s).toJavaObject(OrderOrigin.class);
            return order;
        });

        DataStream<String> jsonString = singleOutputStreamOperator.flatMap((new FlatMapFunction<OrderOrigin, String>() {
            @Override
            public void flatMap(OrderOrigin orderOrigin, Collector<String> collector) throws Exception {
                orderOrigin.getProducts().forEach(product -> {
                    OrderNew orderNew = new OrderNew();
                    orderNew.setOrderId(orderOrigin.getOrderId());
                    orderNew.setUserId(orderOrigin.getUserId());
                    orderNew.setOrderStatus(orderOrigin.getOrderStatus());
                    orderNew.setOrderStatusId(orderOrigin.getOrderStatusId());
                    orderNew.setPayment(orderOrigin.getPayment());
                    orderNew.setPayMode(orderOrigin.getPayMode());
                    orderNew.setPayModeId(orderOrigin.getPayModeId());
                    orderNew.setTs(orderOrigin.getTs());
                    Product p = new Product();
                    p.setProductId(product.getProductId());
                    p.setProductName(product.getProductName());
                    p.setProductNum(product.getProductNum());
                    p.setPrice(product.getPrice());
                    p.setCategoryid(product.getCategoryid());
                    p.setCatname1(product.getCatname1());
                    p.setCatname2(product.getCatname2());
                    p.setCatname3(product.getCatname3());
                    orderNew.setProduct(p);
                    collector.collect(JSON.toJSONString(orderNew));
                });
            }
        }));

        //shuchu
        String topic2 = "lagoudruid3";
        Properties prop2 = new Properties();
        prop2.setProperty("bootstrap.servers","node51:9092,node52:9092,node53:9092");

        //实例化kafka生产者
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>(topic2,
                new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()),
                prop2,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        jsonString.addSink(producer);
        env.execute("kafka_flink_flatmap");
    }
}
